package com.gugee.tiktok.data.consumer.comment.core.main;

import cn.hutool.core.date.DateUtil;
import com.fasterxml.jackson.databind.JavaType;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.gugee.common.thread.ScheduleTemplate;
import com.gugee.tiktok.common.model.dal.blogger.AwemeCommentRepository;
import com.gugee.tiktok.common.model.dal.blogger.AwemeRepository;
import com.gugee.tiktok.common.model.dal.blogger.BloggerFansPortraitRepository;
import com.gugee.tiktok.common.model.dto.SlimAwemeCommentDto;
import com.gugee.tiktok.common.model.es.Keyword;
import com.gugee.tiktok.common.model.mysql.Aweme;
import com.gugee.tiktok.common.model.mysql.AwemeComment;
import com.gugee.tiktok.common.model.spider.comment.AwemeCommentResponse;
import com.gugee.tiktok.common.model.spider.comment.Comment;
import com.gugee.tiktok.data.common.msg.AwemeMessage;
import com.gugee.tiktok.data.common.msg.CommentMessage;
import com.gugee.tiktok.data.common.msg.FansPortraitMessage;
import com.gugee.tiktok.data.common.msg.KeywordMessage;
import com.gugee.tiktok.data.consumer.comment.config.CommentAppConfig;
import com.gugee.tiktok.data.consumer.comment.core.spider.AwemeCommentSpiderClient;
import com.xiguaji.boot.toolkit.JsonUtil;
import com.xiguaji.boot.toolkit.ThreadPoolUtils;

import lombok.extern.slf4j.Slf4j;
import lombok.val;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.search.fetch.subphase.FetchSourceContext;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import javax.annotation.PostConstruct;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.ForkJoinPool;

@Component
@Slf4j
public class SaveAwemeCommentMain   {
    private final String AWEME_INDEX = "aweme_index_";


    @Autowired
    AwemeCommentSpiderClient awemeCommentSpiderClient;

    @Autowired
    AwemeCommentRepository awemeCommentRepository;

    @Autowired
    AwemeComment.AwemeCommentMapper awemeCommentMapper;

    @Autowired
    ScheduleTemplate scheduleTemplate;

    @Autowired
    RestHighLevelClient highLevelClient;
    private String workName = "comment-worker";

    @Autowired
    CommentAppConfig commentAppConfig;
    @Autowired
    KafkaTemplate<Long, KeywordMessage> kafkaTemplateKeyword;
    @Autowired
    KafkaTemplate<Long, CommentMessage> kafkaTemplateComment;

    @Autowired
    KafkaTemplate<Long, FansPortraitMessage> kafkaTemplateFansPortrait;

    @Autowired
    KafkaTemplate<Long, AwemeMessage> kafkaTemplateAweme;

    @Autowired
    SlimAwemeCommentDto.SlimAwemeCommentMapper slimAwemeCommentMapper;

    @Autowired
    BloggerFansPortraitRepository bloggerFansPortraitRepository;

    @Autowired
    AwemeRepository awemeRepository;
    ForkJoinPool forkJoinPool;
    private static final ObjectMapper MAPPER = new ObjectMapper();
    @PostConstruct
    public void init() {
        forkJoinPool = ThreadPoolUtils.forkJoinPool(commentAppConfig.getCommentPollSize(), workName);
        scheduleTemplate.attach(workName, forkJoinPool);
    }
    public void handle(List<CommentMessage> msgs)   {
        try {
            /**
             * todo 从库表批量查询，二次过滤掉不需采集的视频评论,判断条件可与消息属性解耦
             */
        forkJoinPool.submit(() -> msgs.parallelStream().forEach(msg -> {
            try {
                //视频已有评论数小于最低要求则先不爬取
                if (msg.getOCommentCount() < commentAppConfig.getCollectMinComments()) {
                    return;
                }
                //oss中存的评论数

                int s3Comments = msg.getSynComments();
                //视频已收录评论数大于1000则不再爬取
                if (s3Comments >= commentAppConfig.getCollectMaxComments()) {
                    return;
                }

                //评论数已采满，暂时用ossComments代替syn_comments_with_reply，等消息消费完
//                if (s3Comments==msg.getOCommentCount()) {
//                    return;
//                }
                if(msg.getId()==null || msg.getId()==0){
                    log.error("接收到的消息无id值，OAwemeId={}",msg.getOAwemeId());
                    return;
                }
                //开始采集评论
                runSpiderComment(msg);

            } catch (Exception ex) {
                log.error("Comment error. awemeId:{},{}" , msg.getOAwemeId(),ex.getMessage());
            }
        })).get();
        } catch (Exception ex) {
            log.error("Comment thread error.  {}" ,  ex.getMessage());
        }
    }
    private void retryMessage(CommentMessage msg,long awemeId,String retryCause){
        log.error("{},awemeId={},将投递重试={} ",retryCause, awemeId,  msg.getRetryTimes());
        if(msg.getRetryTimes()>0){
            msg.setRetryTimes(msg.getRetryTimes()-1);
            kafkaTemplateComment.send(commentAppConfig.getCommentTopicOnline(),msg);
        }else{
            log.error("重试{}次后依旧失败. awemeId:{}" ,msg.getRetryLimitTimes(), msg.getOAwemeId());
        }
    }
    private void runSpiderComment(CommentMessage msg) {
        long awemeId = msg.getOAwemeId();

        long start = System.currentTimeMillis();
        String preErroeMsg = "评论接口异常:";
        String retryCause = "COMMENTS_NUMS_LESS";
        int nums = 40;
        AwemeCommentResponse response;
        boolean isBreak = false;
        try {
            response = awemeCommentSpiderClient.get(awemeId, 0, nums);
        } catch (Exception e) {
            retryCause = e.getMessage();
            retryMessage(msg, awemeId, preErroeMsg + retryCause);
            return;
        }
        List<Comment> comments = response.getComments();
        //第一次就返回为空评论，可能视频被删或设置隐私不可见，需二次确认
        if (CollectionUtils.isEmpty(comments) && !response.isHas_more()) {
            retryCause = "RETURN_NULL";
            log.error("{},awemeId={},code={}，视频可能被删或设置隐私不可见 ", preErroeMsg + retryCause, awemeId, response.getStatus_code());
//           retryMessage(msg,awemeId,preErroeMsg+retryCause);

            //发送确认视频存在消息
            if (response.getStatus_code() == 0) {
                sendAwemeExistMsg(msg);
            }
            return;
        }

        List<AwemeComment> awemeComments = awemeCommentMapper.mapper(comments);
        //最多CollectMaxComments条
        int needComments = Math.min(msg.getOCommentCount(), commentAppConfig.getCollectMaxComments());
        while (awemeComments.size() <= needComments && response.isHas_more()) {
            try {
                long start2 = System.currentTimeMillis();
                response = awemeCommentSpiderClient.get(awemeId, response.getCursor(), nums);

                if ((System.currentTimeMillis() - start2) / 1000 > commentAppConfig.getDebugTimeLimit()) {
                    log.info("SaveAwemeComment awemeCommentSpiderClient awemeId:{} cursor:{} use time:{}s", awemeId, response.getCursor(), (System.currentTimeMillis() - start2) / 1000);
                }
                List<Comment> tmp = response.getComments();
                if (CollectionUtils.isEmpty(tmp)) {
//                    retryCause=retryCause+"-"+"爬取"+awemeComments.size()+"个后comment返回空";
//                    isBreak=true;
//                    break;
                    continue;
                }
                awemeComments.addAll(awemeCommentMapper.mapper(tmp));
            } catch (Exception e) {
//                log.error("评论接口异常,awemeId={} : {} ", awemeId,  e.getMessage());
                retryCause = e.getMessage();
                isBreak = true;
                retryMessage(msg, awemeId, preErroeMsg + retryCause);
                return;
            }
        }
        long s3Start = System.currentTimeMillis();
        int total = totalComments(awemeComments);
        if (awemeComments.size() < msg.getSynComments()) {
            //缺失太多，经过核实，评论确实就剩这些了
//            if (msg.getSynComments() - awemeComments.size() > commentAppConfig.getCollectStepComments()) {
//                retryCause = retryCause + "-" + "awemeComments.size()=" + awemeComments.size() + "-totalComments=" + total + "-needComments=" + needComments + "-SynComments=" + msg.getSynComments();
//                retryMessage(msg, awemeId, preErroeMsg + retryCause);
//                return;
//            } else {
                //两次采集数量偏差不大，可认为是采集成功，但是只更新采集时间
                log.warn(awemeId+"-比上次采集数少-" + "awemeComments.size()=" + awemeComments.size() + "-totalComments=" + total + "-needComments=" + needComments + "-SynComments=" + msg.getSynComments());
                awemeRepository.updateAwemeCommentClTimeByoAwemeId(System.currentTimeMillis() / 1000, msg.getOAwemeId());

//            }
        } else {
            //todo 保存到oss 需批量优化
            awemeCommentRepository.saveWithTime(awemeId,msg.getOCreateTime(), awemeComments);
            //todo 保存到数据库 需批量优化
            awemeRepository.updateAwemeSynCommentsByoAwemeId(System.currentTimeMillis() / 1000, "comment collect", awemeComments.size(), total, System.currentTimeMillis() / 1000, msg.getOAwemeId());

            //投递关键词消息
//            sendKeyWordMsg2Kafka(awemeId, awemeComments, msg);
            //投递粉丝画像消息
            sendFansPortritMsg(awemeId, awemeComments, msg);


        }
        printStatics(msg,awemeComments.size(),total);
        if ((System.currentTimeMillis() - start) / 1000 > commentAppConfig.getDebugTimeLimit()) {
            log.info("SaveAwemeCommentEvent end awemeId :{} awemeComentSize:{} use :{}s", awemeId, awemeComments.size(), (System.currentTimeMillis() - start) / 1000);
        }
    }
    //日志打印，用于统计采集成功率
   private void printStatics(CommentMessage msg,int clComments,int clCommentsWithReply){
        if (clComments > commentAppConfig.getCollectMinComments()) {
            if(msg.getRetryTimes()< msg.getRetryLimitTimes()){
                log.info(" 重试{}次后成功:AwemeId={}" ,msg.getRetryLimitTimes()-msg.getRetryTimes(),msg.getOAwemeId());
            }else{
                log.info(" 首次采集成功:AwemeId={}" ,msg.getOAwemeId());
            }
        } else {
            if(msg.getRetryTimes()== msg.getRetryLimitTimes()){
                log.info("首次采集到评论数不够要求的{}条,awemeId={},实际采集={}条，含reply统计={}条，应该采集-含reply={}条", commentAppConfig.getCollectMinComments(), msg.getOAwemeId(), clComments,clCommentsWithReply,msg.getOCommentCount());
            }else{
                log.info("重试{}次采集到评论数不够要求的{}条,awemeId={},实际采集={}条，含reply统计={}，条应该采集-含reply={}条",msg.getRetryLimitTimes()-msg.getRetryTimes(), commentAppConfig.getCollectMinComments(), msg.getOAwemeId(),clComments,clCommentsWithReply,msg.getOCommentCount());

            }
        }
    }
    private int totalComments(List<AwemeComment> comments){
        int replyCommentTotal=0;
        if(comments==null || comments.size()==0){
            return 0;
        }
        for(AwemeComment comment:comments){
            replyCommentTotal=replyCommentTotal+comment.getReplyCommentTotal();
        }
        return replyCommentTotal+comments.size();
    }
    private List<AwemeComment> getCommentsFromS3(long awemeId) {
        long start = System.currentTimeMillis();
        List<AwemeComment> result = new ArrayList<>();

        try {
            String comment = "";
            val aweme = awemeRepository.findByoAwemeId(awemeId);
            if(aweme.isPresent()){
                comment = awemeCommentRepository.get(awemeId,aweme.get().getDatecode());
            }

            if ((System.currentTimeMillis() - start) / 1000 > 5) {
                log.info("SaveAwemeComment getCommentCount from oss data awemeId:{} use time:{}s", awemeId, (System.currentTimeMillis() - start) / 1000);
           }

            JavaType jt = MAPPER.getTypeFactory().constructParametricType(ArrayList.class, AwemeComment.class);
            List<AwemeComment> rst = MAPPER.readValue(comment, jt);
            return rst;

        } catch (Exception e) {
             log.error("SaveAwemeComment get oss data error!",e);
        }
        return result;
    }

    private int getCommentCount(long awemeId) {
        long start = System.currentTimeMillis();
        int result = 0;

        try {
            String comment = "";
            val aweme = awemeRepository.findByoAwemeId(awemeId);
            if(aweme.isPresent()){
                comment = awemeCommentRepository.get(awemeId,aweme.get().getDatecode());
            }

            if ((System.currentTimeMillis() - start) / 1000 > 5) {
                log.info("SaveAwemeComment getCommentCount from oss data awemeId:{} use time:{}s", awemeId, (System.currentTimeMillis() - start) / 1000);
            }
            result = JsonUtil.jsonParse(List.class, comment).size();
        } catch (Exception e) {
             log.error("SaveAwemeComment get oss data error!",e);
        }
        return result;
    }


    private List<Keyword> findKeywordByES(Aweme aweme) {
        GetRequest getRequest = new GetRequest(AWEME_INDEX + DateUtil.format(DateUtil.calendar(aweme.getOCreateTime() * 1000).getTime(), "yyyyMM"),
                String.valueOf(aweme.getId()));
        FetchSourceContext fetchSourceContext = new FetchSourceContext(true);
        getRequest.fetchSourceContext(fetchSourceContext);
        GetResponse getResponse;

        try {
            long start = System.currentTimeMillis();
            getResponse = highLevelClient.get(getRequest, RequestOptions.DEFAULT);

            if ((System.currentTimeMillis() - start) / 1000 > 5) {
                log.info("SaveAwemeComment findKeywordByEs awemeId:{} use time:{}ms", aweme.getOAwemeId(), (System.currentTimeMillis() - start));
           }


        } catch (IOException e) {
            log.error("ESRefreshAwemeEventListener findEs error!", e);
            return new ArrayList<>();
        }

        if (null == getResponse.getSource()) {
            return new ArrayList<>();
        }

        if (null == getResponse.getSource().get("awemeTags")) {
            return new ArrayList<>();
        }

        return (List<Keyword>) getResponse.getSource().get("awemeTags");
    }

    /**
     * 投递关键词消息
     * @param awemeId
     * @param awemeComments
     * @param msg
     */
    private void sendKeyWordMsg2Kafka(long awemeId,List<AwemeComment>awemeComments,CommentMessage msg){
        KeywordMessage keywordMessage=new KeywordMessage();
        keywordMessage.setOAwemeId(awemeId);
        keywordMessage.setId(msg.getId());
        keywordMessage.setAwemeComments(awemeComments);
        keywordMessage.setOCreateTime(msg.getOCreateTime());
        keywordMessage.setRetryTimes(commentAppConfig.getRetryKeywordTimes());
        keywordMessage.setRetryLimitTimes(commentAppConfig.getRetryKeywordTimes());
        kafkaTemplateKeyword.send(commentAppConfig.getKeywordTopicOnline(),keywordMessage);
    }

    /**
     * 投递粉丝画像消息
     * @param awemeId
     * @param awemeComments
     * @param msg
     */
    private void sendFansPortritMsg(long awemeId,List<AwemeComment>awemeComments,CommentMessage msg){
        //投递粉丝画像消息所需最小采集评论数
        int minCommentCount = commentAppConfig.getFansPortraitCommentCountIncreaseMin() + msg.getSynComments();
        //视频粉丝画像最近更新时间
        long lastUpdateTime = (System.currentTimeMillis() / 1000 - commentAppConfig.getFansPortraitIntervalHours() * 3600);

        //查询是否满足需要更新的记录
//        Optional<BloggerFansPortrait> bloggerFansPortrait = bloggerFansPortraitRepository.findByAwemeIdAndUpdateTimeAndNeedAnalyzeBloggerCount(awemeId,
//                lastUpdateTime,commentAppConfig.getFansPortraitDisabilityRate());

        //限制评论数超过过去的一定数量时再进行更新，如果 PortraitCommentCountIncreaseMin 为 0 不开启验证
        boolean isNeedSendMsg = commentAppConfig.getFansPortraitCommentCountIncreaseMin() == 0
                || awemeComments.size() >= minCommentCount
                || bloggerFansPortraitRepository.findByAwemeIdAndUpdateTimeAndNeedAnalyzeBloggerCount(awemeId,
                lastUpdateTime,commentAppConfig.getFansPortraitDisabilityRate()).isEmpty();
        if(isNeedSendMsg){
            FansPortraitMessage fansPortraitMessage = new FansPortraitMessage();
            fansPortraitMessage.setUid(msg.getUid());
            fansPortraitMessage.setAwemeId(awemeId);
            fansPortraitMessage.setComments(slimAwemeCommentMapper.mapper(awemeComments));
            fansPortraitMessage.setId(msg.getId());
            fansPortraitMessage.setRetryTimes(commentAppConfig.getRetryFansPortraitTimes());
            fansPortraitMessage.setRetryLimitTimes(commentAppConfig.getRetryFansPortraitTimes());
            kafkaTemplateFansPortrait.send(commentAppConfig.getFansPortraitTopicOnline(),fansPortraitMessage);
        }else{
            log.info("采集到评论数不够要求的增长{}条，不投递粉丝画像消息,awemeId={},实际采集={}条，至少应该采集={}条",
                    commentAppConfig.getFansPortraitCommentCountIncreaseMin(), awemeId, awemeComments.size(),minCommentCount);
        }
    }

    /**
     * 投递确认视频存在消息
     * @param msg
     */
    private void sendAwemeExistMsg(CommentMessage msg){
        AwemeMessage awemeMessage = new AwemeMessage();
        awemeMessage.setId(msg.getId());
        awemeMessage.setAwemeId(msg.getOAwemeId());
        awemeMessage.setRetryTimes(commentAppConfig.getRetryAwemeTimes());
        awemeMessage.setRetryLimitTimes(commentAppConfig.getRetryAwemeTimes());
        kafkaTemplateAweme.send(commentAppConfig.getAwemeExistTopicOnline(),awemeMessage);
    }
}
